/*********************************************************
          File Name: loop_per_thread.cpp
          Author: Abby Cin
          Mail: abbytsing@gmail.com
          Created Time: Mon 09 Oct 2017 10:32:00 AM CST
**********************************************************/

#include <iostream>
#include <thread>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/epoll.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <unistd.h>
#include <stdlib.h>
#include <logging.h>
#include <channel.h>
#include <signal.h>

volatile bool ok{true};

void handler(int sig)
{
  log_info() << "receive signal: " << strsignal(sig);
  ok = false;
}

void thread_func(Receiver<int>& rcv, int wake)
{
  int g_efd = epoll_create(233);
  struct epoll_event rev[255]{};
  char buf[256] = {0};
  int nbytes = 0;

  if(g_efd == -1)
  {
    log_err() << "epoll_create: " << strerror(errno);
  }
  int res = 0;
  int fd;
  struct epoll_event tmp_ev{};
  tmp_ev.data.fd = wake;
  tmp_ev.events = EPOLLIN;
  epoll_ctl(g_efd, EPOLL_CTL_ADD, wake, &tmp_ev);

  while(ok)
  {
    res = epoll_wait(g_efd, rev, sizeof(rev), -1);
    if(res == -1)
    {
      log_err() << "epoll_wait: " << strerror(errno);
      break;
    }
    memset(buf, 0, sizeof(buf));
    for(int i = 0; i < res; ++i)
    {
      if((rev[i].events & EPOLLIN) && !(rev[i].events & EPOLLHUP || rev[i].events & EPOLLERR))
      {
        if(rev[i].data.fd == wake)
        {
          read(wake, &fd, sizeof(fd));
          while(rcv.try_recv(fd))
          {
            log_info() << "received fd: " << fd;
            if(fd < 0)
            {
              break;
            }
            tmp_ev.data.fd = fd;
            tmp_ev.events = EPOLLIN;
            epoll_ctl(g_efd, EPOLL_CTL_ADD, fd, &tmp_ev);
          }
          continue;
        }
        nbytes = read(rev[i].data.fd, buf, sizeof(buf));
        if(nbytes < 0)
        {
          if(errno != EAGAIN || errno != EWOULDBLOCK)
          {
            log_err() << "read: " << strerror(errno);
            close(rev[i].data.fd);
          }
          continue;
        }
        else if(nbytes == 0)
        {
          log_info() << "client " << rev[i].data.fd << " closed connection";
          close(rev[i].data.fd);
          continue;
        }
        nbytes = write(rev[i].data.fd, buf, nbytes);
        if(nbytes == -1)
        {
          if(errno != EAGAIN || errno != EWOULDBLOCK)
          {
            log_err() << "write: " << strerror(errno);
            close(rev[i].data.fd);
          }
          continue;
        }
      }
      else
      {
        log_err() << "unkown revent";
        close(rev[i].data.fd);
      }
    }
  }
  log_info() << "loop thread exit";
}

struct Join
{
  Join(std::thread& t)
    : t_{t}
  {}

  ~Join()
  {
    try{
      t_.join();
    } catch(const std::system_error& e)
    {
      log_err() << "join thread: " << e.what();
    }
  }
  std::thread& t_;
};

int main()
{
  setenv("LOG_TYPE", "custom", 1);
  nm::Logger::create_async();
  struct epoll_event tmp_ev{};
  struct epoll_event rev[233]{};
  int on = 1;
  struct sockaddr_in addr{};
  addr.sin_addr.s_addr = htonl(INADDR_ANY);
  addr.sin_port = htons(8888);
  addr.sin_family = AF_INET;
  auto real_addr = (struct sockaddr*)&addr;

  int efd = epoll_create(233);
  int listenfd = socket(AF_INET, SOCK_STREAM, 0);
  auto [tx, rx] = channel<int>();

  struct sigaction sa{};
  sa.sa_flags = 0;
  sa.sa_handler = handler;
  sigemptyset(&sa.sa_mask);
  sigaction(SIGINT, &sa, NULL);
  std::thread t;
  Join join{t};
  int pair[2] = {-1};

  if(-1 == socketpair(AF_UNIX, SOCK_STREAM, 0, pair))
  {
    log_err() << "socketpair: " << strerror(errno);
    return 1;
  }

  tmp_ev.data.fd = listenfd;
  tmp_ev.events = EPOLLIN;
  epoll_ctl(efd, EPOLL_CTL_ADD, listenfd, &tmp_ev);

  t = std::thread{thread_func, std::ref(rx), pair[1]};

  if(efd == -1)
  {
    log_err() << "epoll_create: " << strerror(errno);
    return 1;
  }

  if(listenfd == -1)
  {
    log_err() << "socket: " << strerror(errno);
    return 1;
  }
  setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
  if(bind(listenfd, real_addr, sizeof(addr)) == -1)
  {
    log_err() << "bind: " << strerror(errno);
    return 1;
  }
  if(listen(listenfd, SOMAXCONN) == -1)
  {
    log_err() << "listen: " << strerror(errno);
    return 1;
  }

  while(ok)
  {
    int res = epoll_wait(efd, rev, sizeof(rev), -1);
    if(res == -1)
    {
      log_err() << "epoll_wait: " << strerror(errno);
      write(pair[0], &res, sizeof(res));
      break;
    }
    for(int i = 0; i < res; ++i)
    {
      if(rev[i].data.fd == listenfd)
      {
        int conn = accept4(listenfd, NULL, NULL, SOCK_NONBLOCK);
        if(conn == -1)
        {
          log_err() << "accept: " << strerror(errno);
          continue;
        }
        log_info() << "send " << conn << " to loop";
        tx.send(conn);
        write(pair[0], &conn, sizeof(conn));
      }
    }
  }
}
